package com.atguigu.extra;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 求窗口内topN的值
 * 求解思路
 * 1、keyby之后 window开窗
 * 2、再根据itemId和窗口进行分组
 * 3、对每个窗口内的元素进行over开窗，进行一个排序
 * 4、将结果写入到mysql中
 *
 */
public class FlinkTopNDemo {
    public static void main(String[] args) {


        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(1);


        //1、创建执行环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        //2、建立一个动态表和文件相关联
        tEnv.executeSql(" create table ub(" +
                " userId bigint, " +
                " itemId bigint, " +
                " categoryId int, " +
                " behavior string, " +
                " ts bigint, " +
                " et as TO_TIMESTAMP_LTZ(ts * 1000, 3)," +       //这里是由ts做一个时间对象出来进行使用
                " watermark for et as et - interval '3' second " +    //这里是添加水印并设置最大乱序时间
                ")with(" +
                " 'connector' = 'filesystem', " +
                " 'path' = 'input/UserBehavior.csv', " +
                " 'format' = 'csv' " +
                ")");

        //tEnv.sqlQuery("select * from ub").execute().print();

        //3、按照商品Id和窗口进行聚合
        Table t1 = tEnv.sqlQuery("select " +
                " itemId, " +
                " hop_start(et, interval '30' minute, interval '1' hour) stt, " +
                " hop_end(et, interval '30' minute, interval '1' hour) edt, " +
                " count(*) ct " +
                " from ub " +
                " where behavior ='pv' " +
                " group by itemId,hop(et, interval '30' minute, interval '1' hour)");
        //.execute().print();

        //4、接下来对每个分组中的元素进行开窗排序,计算名次
        //先注册一下t1表
        tEnv.createTemporaryView("t1",t1);

        Table t2 = tEnv.sqlQuery("select " +
                " itemId, stt, edt, ct, " +
                " row_number() over(partition by edt order by ct desc) rn" +
                " from t1 ");

        //.execute().print();

        //执行到这里的时候会报错
        /*
        Exception in thread "main" org.apache.flink.table.api.TableException: The window can only be ordered in ASCENDING mode.
	at org.apache.flink.table.planner.plan.nodes.exec.stream.StreamExecOverAggregate.translateToPlanInternal(StreamExecOverAggregate.java:131)

        该错误表示当我们在使用开窗函数的时候只能对时间进行升序，除此之外就没有其他的排序方式了
         */

        //5、使用Blink 计划器支持 Top-N的方案继续写

        //注册t2表
        tEnv.createTemporaryView("t2",t2);
        Table result = tEnv.sqlQuery("select edt,itemId,ct,rn from t2 " +
                " where rn <= 3");

        //这时候将结果写入到mysql中去，
        //需要注意的是，如果想将结果写入到数据库中，那么该数据库必须支持覆写操作！

        //6、创建表关联mysql数据库
        tEnv.executeSql("create table TopNProduct(" +
                " w_end timestamp, " +
                " item_id bigint, " +
                " item_count bigint," +
                " rk bigint," +
                " PRIMARY KEY (w_end, rk) NOT ENFORCED " +
                ")with(" +
                "  'connector' = 'jdbc'," +
                "   'url' = 'jdbc:mysql://localhost:3306/flink_sql', " +
                "   'table-name' = 'hot_item'," +
                " 'username' = 'root', " +
                " 'password' = '123321' " +
                " )");

        /*

        且在关联mysql中的表设置主键的时候取消主键检测  " PRIMARY KEY (w_end, rk) NOT ENFORCED "
        踩坑问题：关联mysql中的数据表的时候需要和mysql中创建表的字段名称，类型和顺序保持一致
         */

        //将结果写入到mysql中
        result.executeInsert("TopNProduct");



    }
}
